package com.xblog.chatbox.vlmp.core.message;

import com.xblog.chatbox.vlmp.conversation.*;
import com.xblog.chatbox.vlmp.core.VlmpAgentClient;
import com.xblog.chatbox.vlmp.core.VlmpAgentRequest;
import com.xblog.chatbox.vlmp.core.VlmpAgentResponse;
import com.xiaowu.xblog.constant.AuthorizationContext;
import com.xiaowu.xblog.constant.UserContext;
import com.xiaowu.xblog.utils.SnowflakeIdWorker;
import org.apache.tomcat.util.http.parser.Authorization;
import org.springframework.util.StringUtils;
import reactor.core.publisher.Flux;

import java.util.*;
import java.util.Collections;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @description:
 * @author: xiaowu
 * @time: 2025/8/6 0:14
 */
public abstract class AbstractVlmpAgentMessageProvider implements VlmpAgentMessageProvider{

    private final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(AbstractVlmpAgentMessageProvider.class);

    private final VlmpAgentClient client;

    private final ConversationManager conversationManager;

    public AbstractVlmpAgentMessageProvider(VlmpAgentClient client, ConversationManager conversationManager) {
        this.client = client;
        this.conversationManager = conversationManager;
    }

    private Flux<ConversationResponse> completions(UserContext userContext,VlmpAgentRequest agentRequest,Runnable onError){
        return Flux.defer(() -> {
            if (agentRequest.isStreaming()) {
                return client.stream(userContext, agentRequest).content();
            }
            return client.call(userContext, agentRequest).content().map(chat -> {
                        return Flux.just(chat, VlmpAgentResponse.end(agentRequest));
                    }).orElse(Flux.empty());
        }) //发生错误的时候插入一个错误元素
                .onErrorResume(e -> {
                    onError.run();
                    return Flux.just(VlmpAgentResponse.error(agentRequest,e.getMessage()));
                })
                .map(this::onMessage);
    }

    private ConversationResponse onMessage(VlmpAgentResponse chat) {
        return ConversationResponse.builder()
                .requestId(chat.getRequestId())
                .messageId(chat.getMessageId())
                .conversationId(chat.getConversationId())
                .type(chat.getType().name())
                .payload(chat.getPayload())
                .isEnd(chat.isEnd())
                .build();
    }


    @Override
    public Flux<ConversationResponse> subscribe(UserContext userContext, ConversationRequest request) {
        //对话id
        String conversationId = (String)request.getParameter().get("conversationId");
        //提示词
        String prompt = (String) request.getParameter().get("prompt");
        //是否开启输出流模式
        boolean streaming = (boolean) request.getParameter().getOrDefault("stream",true);
        //是否重新回答
        boolean refresh = (boolean) request.getParameter().getOrDefault("refresh", false);
        //创建对话主题
        Conversation conversation = conversationManager.insert(userContext, Conversation.of(this.getTopic(), conversationId, prompt));

        StringBuffer buffer = new StringBuffer();

        //提供了一系列线程安全的原子操作方法，主要用于无锁并发编程
        AtomicBoolean error = new AtomicBoolean(false);
        //构建聊天
        VlmpAgentRequest agentRequest = VlmpAgentRequest.builder()
                .requestId(request.getRequestId())
                .conversationId(conversation.getConversationId())
                .streaming(streaming)
                .user(userContext.getUsername())
                .text(prompt)
                .metadata(Collections.singletonMap("token", userContext.getToken()))
                .build();
        return this.completions(userContext, agentRequest, () -> error.set(true))
                .onErrorContinue((err, v) -> logger.error(err.getMessage(), err))
                .doOnNext(chat -> Optional.ofNullable(chat.getPayload()).ifPresent(buffer::append))
                .doFinally((signalType) -> {
                    if (logger.isDebugEnabled()) {
                        logger.debug("stream ended with signal: {}", signalType);
                    }
                    if (!StringUtils.hasText(agentRequest.getConversationId())) {
                        logger.warn("conversation is null or empty! prompt: {}", prompt);
                        return;
                    }
                    //重命名对话
                    conversation.setConversationId(agentRequest.getConversationId());
                    conversationManager.rename(userContext, conversation);
                    //保存聊天记录
                    ConversationPayload question = ConversationPayload.Q(
                            conversation,
                            this.getTopic(),
                            agentRequest.getRequestId(),
                            prompt
                    );
                    conversationManager.addPayLoad(userContext, question, refresh);
                    ConversationPayload answer = ConversationPayload.A(
                            conversation,
                            this.getTopic(),
                            agentRequest.getRequestId(),
                            agentRequest.getMessageId(),
                            buffer.toString(),
                            !error.get()
                    );
                    conversationManager.addPayLoad(userContext, answer, refresh);
                });
    }

    @Override
    public void delete(UserContext context, Conversation conversation) {
        client.delete(context, conversation);
    }


    @Override
    public void rename(UserContext context, Conversation conversation) {
        client.rename(context, conversation);
    }

    /**
     * 提交反馈
     *
     * @param feedbacks 反馈信息
     */
    @Override
    public void feedbacks(UserContext authorization, ConversationPayloadFeedbacks feedbacks) {
        client.feedbacks(authorization, feedbacks);
    }
}
